package cloud.xiguapi.ubas.analysis.pv;

import cloud.xiguapi.ubas.analysis.hotitems.model.UserBehaviorAscendingTimestampExtractor;
import cloud.xiguapi.ubas.analysis.pv.model.FilePVCountAggregate;
import cloud.xiguapi.ubas.analysis.pv.model.FilePVCountProcess;
import cloud.xiguapi.ubas.analysis.pv.model.FilePVCountResult;
import cloud.xiguapi.ubas.analysis.pv.model.PVResult;
import cloud.xiguapi.ubas.model.PageViewCount;
import cloud.xiguapi.ubas.model.UserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.Random;

import static cloud.xiguapi.ubas.constant.JobNameConstant.FILE_PV;

/**
 * 离线PV统计
 *
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 上午 08:55
 * desc:
 */
@Component
public class FilePVAnalysis {

    @Resource(name = "env")
    private StreamExecutionEnvironment env;


    @Async("filePVAnalysis")
    public void analysis(String path) throws Exception {
        env.setParallelism(8);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStream<String> inputDataStream = env.readTextFile(path);

        // 数据类型转换
        DataStream<UserBehavior> userBehaviorDataStream = inputDataStream.map(line -> {
            String[] fields = line.split(",");
            return UserBehavior.builder()
                    .userId(Long.valueOf(fields[0]))
                    .itemId(Long.valueOf(fields[1]))
                    .categoryId(Long.valueOf(fields[2]))
                    .behavior(fields[3])
                    .timestamp(Long.valueOf(fields[4]))
                    .build();
        }).assignTimestampsAndWatermarks(new UserBehaviorAscendingTimestampExtractor());


        // 并行任务改进, 设计随机key, 解决数据倾斜问题
        SingleOutputStreamOperator<PageViewCount> pvWindowAggregateDataStream = userBehaviorDataStream.filter(data -> "pv".equals(data.getBehavior()))
                .map(new MapFunction<UserBehavior, Tuple2<Integer, Long>>() {
                    @Override
                    public Tuple2<Integer, Long> map(UserBehavior value) {
                        Random random = new Random();
                        return new Tuple2<>(random.nextInt(12), 1L);
                    }
                })
                .keyBy(data -> data.f0)
                .timeWindow(Time.hours(1L))
                .aggregate(new FilePVCountAggregate(), new FilePVCountResult());

        // 将各分区数据汇总起来
        SingleOutputStreamOperator<PVResult> pvResultDataStream = pvWindowAggregateDataStream
                .keyBy(PageViewCount::getWindowEnd)
                // .sum("count");
                // 自定义处理函数, 将所有窗口数据收集齐后再输出, 解决同一窗口多次输出问题
                .process(new FilePVCountProcess());

        pvResultDataStream.print();
        pvResultDataStream.addSink(new FilePVMySqlSink());

        env.execute(FILE_PV.getJobName());
    }

    public static class FilePVMySqlSink extends RichSinkFunction<PVResult> {

        private transient Connection connection;

        private transient PreparedStatement clearStatement;

        private transient PreparedStatement selectStatement;

        private transient PreparedStatement insertStatement;

        private transient PreparedStatement updateStatement;

        private transient boolean isFirst = true;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/ubas?serverTimezone=UTC", "root", "1015");
            clearStatement = connection.prepareStatement("TRUNCATE TABLE file_pv");
            selectStatement = connection.prepareStatement("SELECT 1 FROM file_pv WHERE window_end = ? LIMIT 1");
            insertStatement = connection.prepareStatement("INSERT INTO file_pv(window_end, cnt) VALUES(?, ?)");
            updateStatement = connection.prepareStatement("UPDATE file_pv SET cnt = ? WHERE window_end = ?");
        }

        @Override
        public void close() throws Exception {
            updateStatement.close();
            selectStatement.close();
            clearStatement.close();
            insertStatement.close();
            connection.close();
        }

        @Override
        public void invoke(PVResult value, SinkFunction.Context context) throws Exception {
            if (isFirst) {
                clearStatement.execute();
                isFirst = false;
                return;
            }

            selectStatement.setString(1, value.getWindowEnd());
            boolean next = selectStatement.executeQuery().next();

            if (next) {
                updateStatement.setLong(1, value.getCount());
                updateStatement.setString(2, value.getWindowEnd());
                updateStatement.execute();
            } else {
                insertStatement.setString(1, value.getWindowEnd());
                insertStatement.setLong(2, value.getCount());
                insertStatement.execute();
            }
        }
    }
}
